package actor

import (
	"fmt"
	"go_learn/mlog"
	"reflect"
	"runtime"
)

type ParallelActor struct {
	BaseActor
}

func (slf *ParallelActor) OnInit(app App, s Service, addr uint) {
	slf.BaseActor.OnInit(app, s, addr)
	slf.BaseActor.scheduler = slf.scheduler
}

func (slf *ParallelActor) RegistHandleGO(msgType EMsgType, handle interface{}) {
	fn := runtime.FuncForPC(reflect.ValueOf(handle).Pointer()).Name()
	name := getFunctionName(fn)
	if name == "" {
		panic(fmt.Errorf("can't find function[%v] name!", handle))
	}
	m, ok := slf.msgHandle[msgType]
	if !ok {
		m = make(map[string]*FunctionInfo)
		slf.msgHandle[msgType] = m
	}
	if _, ok := m[name]; ok {
		panic(fmt.Errorf("function %v: already registered", name))
	}
	finfo := &FunctionInfo{
		Function:  reflect.ValueOf(handle),
		FuncType:  reflect.ValueOf(handle).Type(),
		Goroutine: true,
	}

	finfo.InType = []reflect.Type{}
	for i := 0; i < finfo.FuncType.NumIn(); i++ {
		rv := finfo.FuncType.In(i)
		finfo.InType = append(finfo.InType, rv)
	}
	m[name] = finfo
}

func (slf *ParallelActor) dealSignal(sig interface{}) {
	switch msg := sig.(type) {
	case *RpcMsg:
		slf.dispatch(msg)
	case *RetInfo:
		s := slf.rmSession(msg.session)
		if s == nil {
			mlog.Errorf("can't find session[%d]", msg.session)
		} else if s.cb != nil {
			s.cb(msg.ret, msg.err)
		} else {
			s.c <- msg
		}
	case *ForkSignal:
		msg.f(msg.args...)
	}
}

func (slf *ParallelActor) scheduler(closeSig chan bool) {
	defer func() {
		close(slf.closeChan)
		close(slf.lowChan)
		close(slf.highChan)
	}()
	for {
		var sig interface{}
		select {
		case exit := <-slf.closeChan:
			if exit {
				return
			}
		case sig = <-slf.highChan:
		default:
			select {
			case sig = <-slf.highChan:
			case sig = <-slf.lowChan:
			}
		}
		slf.dealSignal(sig)
	}
}
